package org.javaexpert.spaces.masterworker;

import net.jini.core.entry.Entry;
import net.jini.core.entry.UnusableEntryException;
import static net.jini.core.lease.Lease.FOREVER;
import net.jini.core.transaction.Transaction;
import net.jini.core.transaction.TransactionException;
import net.jini.space.JavaSpace;
import static org.javaexpert.spaces.masterworker.TaskStatusData.TaskStatus.*;
import org.javaexpert.spaces.util.CallableInTxn;
import org.javaexpert.spaces.util.RunnableInTxn;
import org.javaexpert.spaces.util.TransactionTemplate;
import sun.reflect.generics.reflectiveObjects.NotImplementedException;

import java.rmi.RemoteException;
import java.util.Date;
import java.util.Iterator;

/**
 * Created by IntelliJ IDEA.
 * User: piotrga
 * Date: Dec 18, 2007
 * Time: 2:28:39 PM
 * To change this template use File | Settings | File Templates.
 */
public class TaskMonitor {
    JavaSpace space;
    TransactionTemplate txnTemplate;
    private int statusLease = TaskBag.WEEK;
    private static final long NOTHING = -1L;
    private static final int MINUTE = 1000 * 60;
    private String monitorId;

    public TaskMonitor(JavaSpace space, TransactionTemplate txnTemplate, String monitorId) {
        this.monitorId = monitorId;
        assert space != null;
        assert txnTemplate != null;
        this.space = space;
        this.txnTemplate = txnTemplate;
    }

    public Iterable<TaskStatusData> taskStatuses() {
        return new Iterable<TaskStatusData>() {
            public Iterator<TaskStatusData> iterator() {
                return new Iterator<TaskStatusData>() {
                    TaskStatusEntry cashedNext;
                    boolean reachedEnd;

                    private void retrieveTail() {
                        try {
                            cashedNext = (TaskStatusEntry) space.readIfExists(getTailTemplate(), null, FOREVER);
                        } catch (Exception e) {
                            throw new RuntimeException(e);
                        }
                    }

                    public boolean hasNext() {
                        getTailIfNotInitialized();
                        return !reachedEnd;
                    }

                    private void getTailIfNotInitialized() {
                        if (cashedNext == null && !reachedEnd) {
                            retrieveTail();
                            reachedEnd = cashedNext == null;
                        }
                    }

                    public TaskStatusData next() {
                        getTailIfNotInitialized();
                        if (reachedEnd) {
                            return null;
                        }
                        TaskStatusData tmp = cashedNext;
                        try {
                            cashedNext = (TaskStatusEntry) space.readIfExists(taskStatusTemplate(cashedNext.previousTaskId), null, FOREVER);
                            reachedEnd = (cashedNext == null);
                        } catch (Exception e) {
                            throw new RuntimeException(e);
                        }
                        return tmp;
                    }

                    public void remove() {
                        throw new NotImplementedException();
                    }
                };
            }
        };
    }

    public void startWork(final long taskId, final String workerId) {
        txnTemplate.runInTxn(new RunnableInTxn() {
            public void run(Transaction txn) throws Exception {
                TaskStatusEntry stat = (TaskStatusEntry) space.take(taskStatusTemplate(taskId), txn, MINUTE);
                assert stat != null;
                space.write(stat.start(workerId), txn, FOREVER);
            }
        });
    }

    public void addNewTaskStatus(Long taskId, Transaction txn) throws RemoteException, UnusableEntryException, TransactionException, InterruptedException {
        addNewTaskStatus(taskId, null, txn);
    }

    public void addNewTaskStatus(final Long taskId, Long groupId, Transaction txn) throws RemoteException, UnusableEntryException, TransactionException, InterruptedException {
        TaskStatusEntry tail = (TaskStatusEntry) space.takeIfExists(getTailTemplate(), txn, FOREVER);
        Long prevId = NOTHING;
        if (tail != null) {
            prevId = tail.taskId;
            tail.nextTaskId = taskId;
            space.write(tail, txn, statusLease);
        }

        space.write(new TaskStatusEntry(taskId, groupId, monitorId, prevId, NOTHING, PENDING, false), txn, statusLease);
    }

    private TaskStatusEntry getTailTemplate() {
        TaskStatusEntry res = new TaskStatusEntry(monitorId);
        res.nextTaskId = NOTHING;
        return res;
    }

    public void consume(final Long taskId, Transaction txn) throws TransactionException, UnusableEntryException, InterruptedException, RemoteException {
//        txnTemplate.runInTxn(new RunnableInTxn() {
//            public void run(Transaction txn) throws Exception {
        TaskStatusEntry stat = (TaskStatusEntry) space.take(taskStatusTemplate(taskId), txn, MINUTE);
        assert stat != null;
        space.write(stat.consume(), txn, FOREVER);
//            }
//        });
    }

    public void succeed(final Long taskId, Transaction txn) throws RemoteException, TransactionException, UnusableEntryException, InterruptedException {
        TaskStatusEntry stat = (TaskStatusEntry) space.take(taskStatusTemplate(taskId), txn, MINUTE);
        assert stat != null;
        space.write(stat.succeed(), txn, FOREVER);
    }

    public TaskStatusData failOrRestart(final Long taskId, final String error, final Integer maxRetryCount) {
        return txnTemplate.runInTxn(new CallableInTxn<TaskStatusData>() {
            public TaskStatusData call(Transaction txn) throws Exception {
                TaskStatusEntry stat = (TaskStatusEntry) space.take(taskStatusTemplate(taskId), txn, MINUTE);
                assert stat != null;
                stat = stat.failed(error, maxRetryCount);
                space.write(stat, txn, FOREVER);
                return stat;
            }
        });
    }

    public TaskStatusData getTaskStatus(Long id) throws RemoteException, UnusableEntryException, TransactionException, InterruptedException {
        return (TaskStatusData) space.readIfExists(taskStatusTemplate(id), null, FOREVER);
    }

    public boolean isFailed(Long taskId) throws RemoteException, UnusableEntryException, TransactionException, InterruptedException {
        return getTaskStatus(taskId).isFailed();
    }

    public Long waitForFinishedTaskForGroup(Long groupId, long timeout) throws TransactionException, UnusableEntryException, InterruptedException, RemoteException {
        TaskStatusEntry template = new TaskStatusEntry(monitorId);
        template.groupId = groupId;
        template.finished = true;
        return ((TaskStatusEntry) space.read(template, null, timeout)).taskId;
    }


    public static class TaskStatusEntry extends TaskStatusData implements Entry {
        public String monitorId;
        public Long nextTaskId;
        public Long previousTaskId;



        public TaskStatusEntry start(String workerId) {
            switch (status) {
                case PENDING:
                    status = IN_PROGRESS;
                    break;
                case AWAITING_RESTART:
                    status = RESTARTING;
                    break;
                default:
                    throw new IllegalStateException(String.format("Invalid operation start in state %s\n%s\n", status, this.toString()));
            }
            this.workerId = workerId;
            this.startTime = new Date();
            return this;
        }

        public TaskStatusEntry succeed() {
            assert status == IN_PROGRESS || status == RESTARTING;
            status = SUCCEDED;
            finishTime = new Date();
            finished = true;
            return this;
        }

        public TaskStatusEntry failed(String error, Integer maxRetryCount) {
            assert status == IN_PROGRESS || status == RESTARTING;
            if (getRetryCount() >= maxRetryCount) {
                status = FAILED;
                finishTime = new Date();
                finished = true;
            } else {
                incrementRetryCount();
                status = AWAITING_RESTART;
            }
            if (comment == null) {
                comment = "";
            }
            comment += String.format("[%tz] Retry[%d]: %s\n", new Date(), retryCount, error);
            return this;
        }

        private Integer incrementRetryCount() {
            return setRetryCount(getRetryCount() + 1);
        }

        private Integer setRetryCount(int retryCount) {
            this.retryCount = retryCount;
            return retryCount;
        }

        private Integer getRetryCount() {
            return retryCount == null ? 0 : retryCount;
        }


        public TaskStatusEntry(Long taskId, Long groupId, String monitorId, Long previousTaskId, Long nextTaskId, TaskStatus status, Boolean finished) {
            this.taskId = taskId;
            this.groupId = groupId;
            this.previousTaskId = previousTaskId;
            this.nextTaskId = nextTaskId;
            this.monitorId = monitorId;
            this.status = status;
            this.finished = finished;

        }

        public TaskStatusEntry(String monitorId) {
            this.monitorId = monitorId;
        }

        public TaskStatusEntry() {
        }

        public TaskStatusEntry(String monitorId, Long taskId) {
            this.taskId = taskId;
            this.monitorId = monitorId;
        }


        public TaskStatusEntry consume() {
            assert finished;
            assert status == SUCCEDED || status == FAILED;
            consumed = true;
            return this;
        }


    }

    private TaskStatusEntry taskStatusTemplate(Long taskId) {
        return new TaskStatusEntry(monitorId, taskId);
    }
}
